package com.xzc.kafka.action.producer;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.errors.RetriableException;
import org.apache.kafka.common.serialization.Serializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

public class _0ProducerTest {

    public static void main(String[] args) throws Exception {
        String topicName = "test-topic1";
        Properties props = new Properties();
        props.put("bootstrap.servers", "es1:19092,es2:19092,es3:19092");
//        props.put("key.serializer", "org.apache.kafka.common.serializtion.StringSerializer");
//        props.put("value.serializer", "org.apache.kafka.common.serializtion.StringSerializer");
        props.put("acks", "-1");
        props.put("retries", 3);
        props.put("batch.size", 323840);
        props.put("linger.ms", 10);
        props.put("buffer.memory", 33554432);
        props.put("max.block.ms", 3000);

        props.put("compression.type", "lz4");

        Serializer<String> keySerl = new StringSerializer();
        Serializer<String> valSerl = new StringSerializer();
        Producer<String, String> producer = new KafkaProducer<>(props, keySerl, valSerl);
        for (int i = 0; i < 100; i++) {
            ProducerRecord<String, String> record = new ProducerRecord<>(topicName, Integer.toString(i), Integer.toString(i));
            sendAndForget(producer, record);
//            sendAndCallback(producer, record);
//            sendAndSync(producer, record);
        }

        producer.close();
    }

    public static void sendAndForget(Producer<String, String> producer, ProducerRecord<String, String> record) {
        producer.send(record);
    }

    public static void sendAndCallback(Producer<String, String> producer, ProducerRecord<String, String> record) {
        producer.send(record, new Callback() {
            @Override
            public void onCompletion(RecordMetadata metadata, Exception exception) {
                if (exception == null) {
                    System.out.println("消息收到了" + metadata.offset());
                } else {
                    if (exception instanceof RetriableException) {
                        System.out.println("消息异常了，但可以重试，重试可能引起消息时序混乱");
                    } else {
                        System.out.println("消息异常了，不可重试");
                    }
                }
            }
        });
    }

    public static void sendAndSync(Producer<String, String> producer, ProducerRecord<String, String> record) {
        try {
            System.out.println("消息发送前" + record.value());
            producer.send(record).get();
            System.out.println("消息发送后" + record.value());
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
    }
}
